//******************************************************************
//******************************************************************
//**********          ANts Peer To Peer Sources        *************
//
// ANts P2P realizes a third generation P2P net. It protects your
// privacy while you are connected and makes you not trackable, hiding
// your identity (ip) and crypting everything you are sending/receiving
// from others.

// Copyright (C) 2004  Roberto Rossi

// This program is free software; you can redistribute it and/or
// modify it under the terms of the GNU General Public License
// as published by the Free Software Foundation; either version 2
// of the License, or (at your option) any later version.

// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.

// You should have received a copy of the GNU General Public License
// along with this program; if not, write to the Free Software
// Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.

package ants.p2p.filesharing;

import java.io.*;
import java.util.*;
import java.beans.*;
import javax.crypto.*;
import ants.p2p.messages.*;
import ants.p2p.security.*;
import ants.p2p.query.*;
import ants.p2p.query.security.*;
import ants.p2p.utils.indexer.*;
import ants.p2p.utils.encoding.*;
import ants.p2p.*;

import org.apache.log4j.*;

public class FilePullMessageProcessor
    extends Thread {
  PullWatchDog watchDog = new PullWatchDog(this);

  static Logger _logger = Logger.getLogger(FilePullMessageProcessor.class.
                                           getName());

  WarriorAnt n;
  FilePullMessage fpm;
  int percentage = 0;
  long startTime = System.currentTimeMillis();

  boolean terminate = false;

  public FilePullMessageProcessor(WarriorAnt n, FilePullMessage fpm) {
    this.n = n;
    this.fpm = fpm;
    this.setPriority(6);
  }

  public FilePullMessage getFilePullMessage() {
    return this.fpm;
  }

  public void run() {
    FileInputStream fis = null;
    watchDog.start();
    try {
      ants.p2p.security.EndpointSecurityManager esm = n.
          getInputSecureConnectionManager(fpm.getSource());
      if (esm == null) {
        SecureConnectionErrorControlMessage fterrcm = new
            SecureConnectionErrorControlMessage(new Integer(0), null,
                                                "No secure connection");
        MessageWrapper wm = n.sendMessage(fterrcm, fpm.getSource(), false, false);
        _logger.debug(n.getShortId() + ": No secure connection with " +
                      fpm.getSource());
        while (n.myMessages.contains(wm) && !this.terminate) {
          Thread.sleep(1000);
        }
        if (n.failedMessages.contains(wm)) {
          n.failedMessages.remove(wm);
          _logger.debug(n.getShortId() +
                        ": Error in sending ControlMessage " +
                        fterrcm.getMessage());
        }
        watchDog.terminate();
        throw new Exception("No secure connection avaiable with endpoint " +
                            fpm.getSource());
      }

      try {
        fpm.decrypt(esm.getCipherDec());
      }
      catch (Exception e) {
        SecureConnectionErrorControlMessage fterrcm = new
            SecureConnectionErrorControlMessage(new Integer(0), null,
                                                "No secure connection");
        MessageWrapper wm = n.sendMessage(fterrcm, fpm.getSource(), false, false);
        _logger.debug(n.getShortId() + ": No secure connection with " +
                      fpm.getSource());
        while (n.myMessages.contains(wm) && !this.terminate) {
          Thread.sleep(1000);
        }
        if (n.failedMessages.contains(wm)) {
          n.failedMessages.remove(wm);
          _logger.debug(n.getShortId() +
                        ": Error in sending ControlMessage " +
                        fterrcm.getMessage());
        }
        watchDog.terminate();
        throw new Exception("No secure connection avaiable with endpoint " +
                            fpm.getSource(), e);
      }

      if (n.checkInServiceFilePullRequests(false, fpm)) {
        watchDog.terminate();
        return;
      }

      byte[] hash = fpm.getHash();
      Long offset = fpm.getOffset();
      File f = null;

      //controlla ultima modifica e caratteristiche del file
      BackgroundEngine be = BackgroundEngine.getInstance();
      boolean shared = false;
      boolean partial = false;
      String fpmHash = Base16.toHexString(hash);
      FileInfos fileInfo = (FileInfos) be.sharedFilesIndexHash.get(fpmHash);
      if (fileInfo != null) {
        shared = true;
        f = new File(fileInfo.getName());
      }
      if (!shared || f == null || !f.exists()) {
        Object partialFile = BackgroundEngine.getInstance().getPartialFile(
            fpmHash, QueryHashItem.ANTS_HASH);
        if (partialFile != null) {
          shared = true;
          long partId = fpm.getOffset().longValue() /
              (fpm.getBlockSize().intValue() * fpm.getBlocks().longValue());
          f = new File(WarriorAnt.chunksHome + WarriorAnt.chunksPath + fpmHash +
                       "." + partId);
          partial = true;
        }
      }
      if (!shared || f == null || !f.exists()) {
        FileTransferErrorControlMessage fterrcm = new
            FileTransferErrorControlMessage(new Integer(0), hash,
                                            "File don't exist", fpm);
        fterrcm.encrypt(esm.getCipherEnc());
        MessageWrapper wm = n.sendMessage(fterrcm, fpm.getSource(), false, false);
        _logger.debug(n.getShortId() + ": File " + fpm.getFileName() +
                      " don't exist. Cannot serve pull request from id " +
                      fpm.getSource());
        while (n.myMessages.contains(wm) && !this.terminate) {
          Thread.sleep(1000);
        }
        if (n.failedMessages.contains(wm)) {
          n.failedMessages.remove(wm);
          fterrcm.decrypt(esm.getCipherDec());
          _logger.debug(n.getShortId() +
                        ": Error in sending ControlMessage " +
                        fterrcm.getMessage());
        }
        watchDog.terminate();
        return;
      }

      if (n.checkInServiceFilePullRequests(true, fpm)) {
        watchDog.terminate();
        return;
      }

      fis = new FileInputStream(f);
      FilePushMessage frm = new FilePushMessage(fpm.getLocalFileName(), hash,
                                                offset, fpm.getBlockSize(),
                                                fpm.getResume(),
                                                new Long(f.length()));
      frm.encrypt(esm.getCipherEnc());
      MessageWrapper wm = n.sendMessage(frm, fpm.getSource(), false, false);
      while (n.myMessages.contains(wm) && !n.isDisconnected() &&
             !this.terminate) {
        Thread.sleep(1000);
      }
      if (n.failedMessages.contains(wm)) {
        n.failedMessages.remove(wm);
        frm.decrypt(esm.getCipherDec());
        _logger.debug(n.getShortId() +
                      ": Error in sending FilePushMessage " +
                      frm.getFileName() +
                      " - Cannot serve pull request from id " +
                      fpm.getSource());
        n.inServiceFilePullRequests.remove(fpm);
        watchDog.terminate();
        return;
      }
      watchDog.setReceivedPart();

      frm.decrypt(esm.getCipherDec());
      n.propertyChangeSupport.firePropertyChange("filePullInit", null, this);

      byte[] block = new byte[fpm.getBlockSize().intValue()];
      long blocks = fpm.getBlocks().longValue();
      if (!partial) {
        fis.skip(offset.longValue());
      }
      long partId = 0;
      long partsUploaded = 0;
      ArrayList messageGroup = new ArrayList();
      long sizeUploaded = 0;
      while (fis.available() > 0 && partId < blocks && !this.terminate) {
        for (int g = 0;
             g <
             (PartialFile.computeGroupFactor(fpm.getBlockSize().intValue()) -
              messageGroup.size()) && partId < blocks && !this.terminate; g++) {
          int read = fis.read(block);
          if (read > 0) {
            if (read < block.length) {
              byte[] reduced = new byte[read];
              for (int x = 0; x < read; x++) {
                reduced[x] = block[x];
              }
              block = reduced;
            }
            FilePartMessage filepartmess = new FilePartMessage(block, hash, new Long(partId), frm);
            filepartmess.encrypt(esm.getCipherEnc());
            MessageWrapper wm1 = n.sendMessage(filepartmess, fpm.getSource(), false, false);
            sizeUploaded += block.length;
            partId++;
            messageGroup.add(wm1);
            while (messageGroup.size() >= 2 &&
                   !n.areInFailedMessages(messageGroup) && !this.terminate) {
              int lastSize = messageGroup.size();
              partsUploaded += messageGroup.size();
              messageGroup = n.removeDeliveredMessagesFromList(messageGroup);
              int currentSize = messageGroup.size();
              partsUploaded -= messageGroup.size();
              if (lastSize > currentSize) {
                watchDog.setReceivedPart();
              }
              this.percentage = (int) ( ( (partsUploaded * 1.0) / blocks) *
                                       100.0);
              n.propertyChangeSupport.firePropertyChange("filePullUpdate", null, this);
              Thread.sleep(1000);
            }
            if (n.areInFailedMessages(messageGroup)) {
              n.failedMessages.removeAll(messageGroup);

              n.inServiceFilePullRequests.remove(fpm);
              n.propertyChangeSupport.firePropertyChange("filePullError", null,
                  this.getFilePullMessage());
              _logger.info(n.getShortId() + ": Error in file transfer " +
                           fpm.getFileName() +
                           ". Cannot serve pull request from id " +
                           fpm.getSource());
              fis.close();

              watchDog.terminate();
              return;
            }
          }
        }
      }
      while (messageGroup.size() > 0 &&
             !n.areInFailedMessages(messageGroup) && !this.terminate) {
        int lastSize = messageGroup.size();
        partsUploaded += messageGroup.size();
        messageGroup = n.removeDeliveredMessagesFromList(messageGroup);
        int currentSize = messageGroup.size();
        partsUploaded -= messageGroup.size();
        if (lastSize > currentSize) {
          watchDog.setReceivedPart();
        }
        this.percentage = (int) ( ( (partsUploaded * 1.0) / blocks) * 100.0);
        n.propertyChangeSupport.firePropertyChange("filePullUpdate", null, this);
        Thread.sleep(1000);
      }
      if (n.areInFailedMessages(messageGroup)) {
        n.failedMessages.removeAll(messageGroup);

        n.inServiceFilePullRequests.remove(fpm);
        n.propertyChangeSupport.firePropertyChange("filePullError", null,
            this.getFilePullMessage());
        _logger.info(n.getShortId() + ": Error in file transfer " +
                     fpm.getFileName() +
                     ". Cannot serve pull request from id " +
                     fpm.getSource());
        fis.close();

        watchDog.terminate();
        return;
      }

      fis.close();

      n.inServiceFilePullRequests.remove(fpm);
      FileTransferEndControlMessage cm = new FileTransferEndControlMessage(new
          Integer(0), hash, "", frm);
      cm.encrypt(esm.getCipherEnc());
      MessageWrapper wm2 = n.sendMessage(cm, fpm.getSource(), false, false);
      while (n.myMessages.contains(wm2) && !this.terminate) {
        Thread.sleep(1000);
      }
      if (n.failedMessages.contains(wm2)) {
        n.failedMessages.remove(wm2);

        n.propertyChangeSupport.firePropertyChange("filePullError", null,
            this.getFilePullMessage());
        _logger.info(n.getShortId() + ": Error in file transfer " +
                     fpm.getFileName() +
                     ". Cannot serve pull request from id " +
                     fpm.getSource());
        watchDog.terminate();
        return;
      }
      watchDog.setReceivedPart();

      NeighbourAnt.totalUploaded += sizeUploaded;
      n.propertyChangeSupport.firePropertyChange("filePullEnd", null,
                                                 this.getFilePullMessage());
      _logger.info(n.getShortId() + ": File sent " + fpm.getFileName() + " - " +
                   fpm.getLocalFileName() +
                   ". Pull request from id " +
                   fpm.getSource() + " served.");
      watchDog.terminate();
    }
    catch (Exception e) {
      try {
        fis.close();
      }
      catch (Exception ex) {}

      n.inServiceFilePullRequests.remove(fpm);
      n.propertyChangeSupport.firePropertyChange("filePullError", null,
                                                 this.getFilePullMessage());
      watchDog.terminate();

      _logger.error(n.getShortId() +
                    ": Error In Processing FilePullMessage: id = " +
                    fpm.getAck_Id() + " Source: " + fpm.getSource() +
                    " Dest: " + fpm.getDest(), e);
    }
  }

  public void terminate() {
    this.terminate = true;
    this.watchDog.terminate();
    try {
      ants.p2p.security.EndpointSecurityManager esm = n.
          getInputSecureConnectionManager(fpm.getSource());
      if (esm == null) {
        SecureConnectionErrorControlMessage fterrcm = new
            SecureConnectionErrorControlMessage(new Integer(0), null,
                                                "No secure connection");
        MessageWrapper wm = n.sendMessage(fterrcm, fpm.getSource(), false, false);
        _logger.debug(n.getShortId() + ": No secure connection with " +
                      fpm.getSource());
        while (n.myMessages.contains(wm) && !this.terminate) {
          Thread.sleep(1000);
        }
        if (n.failedMessages.contains(wm)) {
          n.failedMessages.remove(wm);
          _logger.debug(n.getShortId() +
                        ": Error in sending ControlMessage " +
                        fterrcm.getMessage());
        }
        watchDog.terminate();
        _logger.error("No secure connection avaiable with endpoint " +
                      fpm.getSource());
        return;
      }

      FileTransferErrorControlMessage cm = new
          FileTransferErrorControlMessage(new
                                          Integer(1), fpm.getHash(), "",
                                          fpm);
      cm.encrypt(esm.getCipherEnc());
      MessageWrapper wm2 = n.sendMessage(cm, fpm.getSource(), false, false);
      while (n.myMessages.contains(wm2)) {
        Thread.sleep(1000);
      }
      if (n.failedMessages.contains(wm2)) {
        n.failedMessages.remove(wm2);
        _logger.info(n.getShortId() + ": Error file transfer interruption " +
                     fpm.getFileName() +
                     ". Cannot notificate interruption of msg id " +
                     fpm.getSource());
        return;
      }
      _logger.warn(n.getShortId() +
                   ": File pull interrupted: id = " +
                   fpm.getAck_Id() + " Source: " + fpm.getSource() +
                   " Dest: " + fpm.getDest());
    }
    catch (Exception ex) {
      _logger.error(
          "Exception while trying to notify file transfer interruption", ex);
    }
  }

  public double getSpeedValue() {
    double elapsed = (System.currentTimeMillis() - this.startTime) / 1000.0;
    double speed = (this.percentage / 100.0) *
        (fpm.getBlocks().longValue() * fpm.getBlockSize().longValue()) /
        (elapsed * Math.pow(2, 10));
    return speed > 0 ? speed : 0;
  }

  public String speedToString() {
    double speed = this.getSpeedValue();
    return ("" + speed).substring(0, ("" + speed).indexOf(".") + 2) + "KiB/s ";
  }

  public String toString() {
    String rv = percentage + "%";
    while (rv.length() < 7) {
      rv += " ";
    }
    rv += this.speedToString();
    while (rv.length() < 15) {
      rv += " ";
    }
    rv += "     " + ji.JI.i("Hash") + ": " + fpm.getLocalFileName();
    while (rv.length() < 80) {
      rv += " ";
    }
    if (this.n.acceptTCPDirectConnections() &&
        !fpm.getSourceAddress().equals("")) {
      rv += ji.JI.i("Requirer") + ": " + fpm.getSourceAddress();
    }
    else {
      rv += ji.JI.i("Requirer") + ": " + fpm.getSource().substring(0, 10);
    }
    rv += "    " + ji.JI.i("Name") + ": " + fpm.getFileName();
    return rv;
  }
}

class PullWatchDog
    extends Thread {
  public static long downloadTimeout = Ant.messageTimeout *
      Ant.maxRetransmissions;
  long lastPartReceivedTime;
  FilePullMessageProcessor fpmp = null;
  boolean interrupt = false;

  static Logger _logger = Logger.getLogger(PullWatchDog.class.getName());

  public PullWatchDog(FilePullMessageProcessor fpmp) {
    super();
    this.fpmp = fpmp;
    lastPartReceivedTime = System.currentTimeMillis();
    this.setPriority(1);
  }

  public void setReceivedPart() {
    lastPartReceivedTime = System.currentTimeMillis();
  }

  public void terminate() {
    this.interrupt = true;
  }

  public void run() {
    while (fpmp.isAlive() && !interrupt) {
      try {
        sleep(5000);
      }
      catch (Exception e) {
        _logger.error("", e);
      }
      long elapsed = System.currentTimeMillis() - this.lastPartReceivedTime;
      if (elapsed > downloadTimeout || fpmp.n.isDisconnected()) {
        fpmp.terminate();
        _logger.info("File pull message processor timed out");
        fpmp.n.propertyChangeSupport.firePropertyChange("filePullError", null,
            fpmp.getFilePullMessage());
        fpmp.n.inServiceFilePullRequests.remove(fpmp.fpm);
        return;
      }
    }
  }
}
